package Yanxin.atguigu.yx.sink;

import Yanxin.atguigu.yx.bean.TableProcess;
import Yanxin.atguigu.yx.util.DruidDSUtil;
import Yanxin.atguigu.yx.util.RedisUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.sql.SQLException;


public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private DruidDataSource druidDataSource;
    private Jedis redisClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取连接池
        druidDataSource = DruidDSUtil.getDruidDataSource();

        redisClient = RedisUtil.getRedisClient();


    }

    @Override
    public void close() throws Exception {
        druidDataSource.close();  // app关闭之前关闭连接池

        redisClient.close();
    }



    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {

        //这是插入维度数据，当数据更新时，要把更新的数据同步到redis中，需要先删除旧的缓存

        //1.向phoenix发送数据
        sendToPhoenix(value);

        //2.删除缓存
        delCache(value);

    }

    private void delCache(Tuple2<JSONObject, TableProcess> value) {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;

        // {id: "1", "sku_name": ''}
        String key = tp.getSinktable() + ":" + data.getString("id");

        //todo 当只有维度更新的时候再去删除,op_type是后面加上去的类型，专门记录维度变化的数据
        if ("update".equals(data.getString("op_type"))) {
            redisClient.del(key);
        }

    }

    private void sendToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        DruidPooledConnection conn = druidDataSource.getConnection();
        JSONObject data = value.f0;
        TableProcess tp = value.f1;

        StringBuilder sql = new StringBuilder();
        sql
                .append("upsert into ")
                .append(tp.getSinktable())
                .append("(")
                .append(tp.getSinkcolumns())
                .append(")values(")
                //
                .append(tp.getSinkcolumns().replaceAll("[^,]+", "?"))
                .append(")");
        System.out.println("插入语句" + sql);


        // 2. 使用连接对象获取一个 PrepareStatement
        PreparedStatement ps = conn.prepareStatement(sql.toString());

        // 3. 给占位符赋值 TODO
        String[] cs = tp.getSinkcolumns().split(",");

        for (int i = 0, count = cs.length; i < count; i++) {

            //todo 这一步就是提取数据，根据对应的列，填入数据
            Object v = data.get(cs[i]);  // null + "" = "null"

            String vv = v == null ? null : v.toString();

            ps.setString(i + 1, vv);
        }

        // 4. 执行sql
        ps.execute();
        conn.commit();
        // 5.关闭PrepareStatement
        ps.close();
        // 6. 归还连接
        conn.close();
    }

}
